package cn.cutie.clot.mq.client;

import cn.cutie.clot.mq.model.ClotMessage;
import cn.cutie.clot.mq.model.Result;
import cn.kimmking.utils.HttpUtils;
import cn.kimmking.utils.ThreadUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.Getter;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import java.util.ArrayList;
import java.util.List;

/**
 * @Description:
 * @Author: Cutie
 * @CreateDate: 2024/6/27 12:28
 * @Version: 0.0.1
 */
public class ClotBroker {

    @Getter
    public static ClotBroker Default = new ClotBroker();

    public static String brokerUrl = "http://localhost:8765/clotmq";

    // topic和mq映射
//    Map<String, ClotMq> mqMapping = new ConcurrentHashMap<>();

//    public ClotMq find(String topic) {
//        return mqMapping.get(topic);
//    }

//    public ClotMq createTopic(String topic) {
//        return mqMapping.putIfAbsent(topic, new ClotMq(topic));
//    }

    static {
        init();
    }

    private static void init() {
        ThreadUtils.getDefault().init(1);
        ThreadUtils.getDefault().schedule(() -> {
            MultiValueMap<String, ClotConsumer<?>> consumers = getDefault().getConsumers();
            consumers.forEach((topic, consumers1) -> {
                consumers1.forEach(consumer -> {
                    ClotMessage<?> recv = consumer.recv(topic);
                    if (recv == null) return;
                    try{
                        consumer.getListener().onMessage(recv);
                        consumer.ack(topic, recv);
                    } catch (Exception e){
                        // TODO: 2024/7/13
                    }

                });
            });
        }, 100, 100);
    }

    public ClotProducer createProducer() {
        return new ClotProducer(this);
    }

    public ClotConsumer<?> createConsumer(String topic) {
        ClotConsumer<?> clotConsumer = new ClotConsumer<>(this);
        clotConsumer.sub(topic);
        return clotConsumer;
    }

    public boolean send(String topic, ClotMessage message) {
        System.out.println("==> send topic/message:" + topic + "/" +message);
        System.out.println("===> send message: " + JSON.toJSONString(message));
        Result<String> result = HttpUtils.httpPost(JSON.toJSONString(message),
                brokerUrl + "/send?t=" + topic, new TypeReference<Result<String>>(){});
        System.out.println("==> send result" + result);
        return result.getCode() == 1;
    }

    public void sub(String topic, String cid) {
        System.out.println("==> send topic/cid:" + topic + "/" +cid);
        Result<String> result = HttpUtils.httpGet(brokerUrl + "/sub?t=" + topic + "&cid=" + cid,
                new TypeReference<Result<String>>(){});
        System.out.println("==> sub result:" + result);
    }

    public <T> ClotMessage<T> recv(String topic, String id) {
        System.out.println("==> recv topic/id:" + topic + "/" +id);
        Result<ClotMessage<String>> result = HttpUtils.httpGet(brokerUrl + "/recv?t=" + topic + "&cid=" + id,
                new TypeReference<Result<ClotMessage<String>>>(){});
        System.out.println("==> recv result:" + result);
        return (ClotMessage<T>) result.getData();
    }

    public void unsub(String topic, String cid) {
        System.out.println("==> unsend topic/cid:" + topic + "/" +cid);
        Result<String> result = HttpUtils.httpGet(brokerUrl + "/unsub?t=" + topic + "&cid=" + cid,
                new TypeReference<Result<String>>(){});
        System.out.println("==> unsend result:" + result);
    }

    public boolean ack(String topic, String cid, int offset) {
        System.out.println("==> ack topic/cid/offset:" + topic + "/" +cid + "/" + offset);
        Result<String> result = HttpUtils.httpGet(brokerUrl + "/ack?t=" + topic + "&cid=" + cid + "&offset=" + offset,
                new TypeReference<Result<String>>(){});
        System.out.println("==> ack result:" + result);
        return result.getCode() == 1;
    }


//    public <T> void addListener(String topic, ClotListener<T> listener) {
//        listeners.add(topic, listener);
//    }

    @Getter
    private MultiValueMap<String, ClotConsumer<?>> consumers = new LinkedMultiValueMap<>();
    public <T> void addConsumer(String topic, ClotConsumer<T> clotConsumer) {
        consumers.add(topic, clotConsumer);
    }
}
